package com.corn.flink.lesson3;

import org.apache.flink.api.common.functions.Partitioner;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * flink 物理分区
 *
 * @author JimWu
 * @date 2023/2/24 15:57
 **/
public class FlinkPhysicalPartitionDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.fromElements(
                new Person("alice", 15, "cq"),
                new Person("jack", 18, "cq"),
                new Person("rose", 20, "sh"),
                new Person("mick", 42, "gz"),
                new Person("white", 12, "cq"),
                new Person("maria", 32, "sz"),
                new Person("candy", 56, "bj"),
                new Person("jerry", 28, "wh")
        )
//                .shuffle() // 随机分区
//                .rebalance() // 轮训分区
//                .rescale() // 重缩放分区
//                .broadcast() // 广播分区
//                .global() // 全局分区 所有的算子会发送到下游算子的第一个并行子任务去
//                .partitionCustom(new Partitioner<Person>() {
//                    /**
//                     * partition
//                     *
//                     * @Param person 输入数据
//                     * @param i 一共几个并行任务 没有0!! 1 - N
//                     * @return int 分区到哪个分区
//                     */
//                    @Override
//                    public int partition(Person person, int i) {
//                        return person.getAge() % 2;
//                    }
//                }, (KeySelector<Person, Person>) person -> person)
                .print().setParallelism(4);

        env.execute();
    }
}
